package com.biubiux2.handler;

import com.biubiux2.async.DisruptorUtil;
import com.biubiux2.async.consumer.entity.MsgEvent;
import com.biubiux2.dao.DaoUtil;
import com.biubiux2.dao.UserMapper;
import com.biubiux2.dialog.DialogContext;
import com.biubiux2.dialog.Order;
import com.biubiux2.entity.User;
import com.biubiux2.session.ServerSession;
import com.biubiux2.session.SessionMap;
import com.lmax.disruptor.RingBuffer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.RejectedExecutionHandler;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author noexcs
 * @since 8/7/2022 3:52 PM
 */
@Slf4j
@Deprecated
public class MessageProcessor {

    private static ExecutorService executor =
            new ThreadPoolExecutor(2,100,1,TimeUnit.MINUTES,new ArrayBlockingQueue<>(1000),new ThreadPoolExecutor.CallerRunsPolicy());

    private static AtomicInteger onlineCnt = new AtomicInteger(0);

    public static LinkedBlockingDeque<Order> deque = new LinkedBlockingDeque<>();

    static {
//        scheduledExecutorService = new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "batch-flush-thread"));
//        scheduledExecutorService.schedule(() -> {
//            DialogContext.currentOnlineUser.forEach((userid, user) -> {
//                user.getChannel().flush();
//            });
//        }, 3, TimeUnit.SECONDS);
        executor.execute(() -> {
            while (true) {
                try {
                    Order order = deque.takeLast();
                    order.channel.writeAndFlush(order.byteBuf);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }

    public void handleLogin(ChannelHandlerContext ctx, DialogContext dialogContext, ByteBuf byteBuf) {
        int userId = byteBuf.readInt();
        int readableBytes = byteBuf.readableBytes();
        byte[] passwordByte = new byte[readableBytes];
        byteBuf.readBytes(passwordByte);
        String password = new String(passwordByte, StandardCharsets.UTF_8);

        executor.execute(() -> {
            UserMapper userMapper = DaoUtil.getMapper(UserMapper.class);
            User user = userMapper.selectPasswordAndUsername(userId);
            if (user != null && user.getPasswd().equals(password)) {
                user.setUserId(userId);
                user.setChannel(ctx.channel());
                dialogContext.channel = ctx.channel();
                dialogContext.userId = userId;
                dialogContext.username = user.getUsername();
                dialogContext.setLoggedIn(true);
                DialogContext.currentOnlineUser.put(userId, user);
                ServerSession session = new ServerSession(ctx.channel());
                session.bindChannel();
                SessionMap.getInstance().addSession(session);
                log.debug("用户{}({})上线", user.getUsername(), userId);
                ByteBuf response = PooledByteBufAllocator.DEFAULT.directBuffer();
                response.writeInt(3);
                response.writeBytes("OK".getBytes(StandardCharsets.UTF_8));
                ctx.channel().writeAndFlush(response);
            } else {
                ctx.channel().writeAndFlush("账号或密码错误".getBytes(StandardCharsets.UTF_8));
            }
        });
    }


    public static void asyncHandleEvent(Channel channel, ByteBuf byteBuf) {
        RingBuffer<MsgEvent> ringBuffer = DisruptorUtil.ringBuffer;
        long nextSeq = ringBuffer.next();
        try {
            MsgEvent msgEvent = ringBuffer.get(nextSeq);
            msgEvent.channel = channel;
            msgEvent.byteBuf = byteBuf;
        } finally {
            ringBuffer.publish(nextSeq);
        }
    }
}
